package com.example.spark.rdd;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.AbstractJavaRDDLike;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class RDDWordCount {

    public static void main(String[] args) {
        //创建配置对象
        SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("WordCount");

        //创建Spark的上下文
        JavaSparkContext jsc = new JavaSparkContext(conf);
        //编写逻辑代码
        //读取input目录下的words.txt文件的数据
        JavaRDD<String> rdd = jsc.textFile("input/words.txt");
        //使用FlatMap转换功能把长字符串切分单个单词
        JavaRDD<String> flatMapRDD = rdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                //line:  a good beginning is half the battle
                String[] words = line.split(" "); //words: {a, good ,beginning, is, half, the, battle }
                List<String> list = Arrays.asList(words);

                return list.iterator();
            }
        });

        //输出测试
        //print(flatMapRDD);

////        //转换格式（word, 1）
        JavaPairRDD<String, Integer> pairRDD = flatMapRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                //Tuple2：[(单词,次数)，(单词,次数)，(单词,次数)]
                return new Tuple2<>(word,1);
            }
        });
////        //输出测试
      //  print(pairRDD);
////        //合并相同的单词 (hello,1),(hello,1) => (hello,2)
        JavaPairRDD<String, Integer> reduceByKeyRDD = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {

                //(hello,1),(hello,1) => (hello,2) 即：1+1=2
                return (Integer) (v1+v2);
            }
        });
        //输出
        print(reduceByKeyRDD);

        //回收资源
        jsc.stop();


    }

    public static void print(AbstractJavaRDDLike rdd){
        rdd.collect().forEach(System.out::println);

    }
}
